//------------------------------------------------------------------------------
// PasCassa - a Delphi/Pascal class library for Apache Cassandra
//
// Copyright (c) 2010-2013 Jens Geyer
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//------------------------------------------------------------------------------
unit ReadWrite.Logic;

interface

uses Classes, Windows, SysUtils, Math, SyncObjs, ComObj, Messages,
     Cassandra_200,
     Cassandra.Client,
     Database.ReadWrite.Client,
     DataModel.Entity.TestData,
     ReadWrite.Constants;

type
  ITestManager = interface
    ['{DE6AEDAC-D369-4794-A1A5-8CE650414C22}']
    function  Name : string;
    procedure Shutdown;
    function  Terminated : Boolean;
    function  CreateDbClient( aReset : Boolean = FALSE) : ITestDataDatabaseClient;
    procedure NotifyErrors( iNr, iErrors : Integer);
    //function  NeueBestellung( const aClient : ITestDataDatabaseClient; aKundenNr : Integer) : IOrder;
  end;


  TTestManager = class( TInterfacedObject, ITestManager)
  protected
    FTerminated : Boolean;
    FName       : string;
    FHandle     : HWND;

    // ITestManager
    function  Name : string;
    procedure Shutdown;
    function  Terminated : Boolean;
    function  CreateDbClient( aReset : Boolean = FALSE) : ITestDataDatabaseClient;
    procedure NotifyErrors( iNr, iErrors : Integer);
    //function  NeueBestellung( const aClient : ITestDataDatabaseClient; aKundenNr : Integer) : IOrder;
  public
    constructor Create( aName : string; aWnd : HWND);
    destructor Destroy;  override;
  end;


  TPerson = class( TThread)
  protected
    FDBClient  : ITestDataDatabaseClient;
    FManager     : ITestManager;
    FNr        : Integer;
    FWaitFix   : WORD;
    FWaitRnd   : WORD;

    procedure PerformTheTask;  virtual; abstract;
    procedure Execute;   override;
    class function CreateGUID : TGuid;

  public
    constructor Create( const aManager : ITestManager; aNr : Integer);
    destructor Destroy;  override;

    property Nr : Integer read FNr;
  end;


  TReadWriteTester = class( TPerson)
  protected
    const READWRITETEST_ROWKEY = 'ReadWrite-Test';
  protected
    FConsistency : TConsistencyLevel;
    FErrors : Integer;
    function  ReadValue( const pData : ITestDataFamily) : Integer;
    procedure WriteValue( const pData : ITestDataFamily; aValue : Integer);

    procedure PerformTheTask;  override;
  public
    constructor Create( const aManager : ITestManager; aNr : Integer; aLevel : TConsistencyLevel);
  end;


const
  WMA_NOTIFYERRORS = WM_APP + 47;

implementation

var
  g_hWakeUp  : THandle = 0;   // Root wakes one or more Actors
  g_evAkteur : TEvent = nil;  // Actor notifies Root about something



//--- Tools --------------------------------------------------------------------

{.$DEFINE Deterministic}

{$IFDEF Deterministic}
var g_notRandom : Integer = 0;

function Random( aLimit : Integer) : Integer;
begin
  result := InterlockedExchangeAdd( g_notRandom, +1) mod aLimit;
end;
{$ENDIF}


//--- TTestManager -------------------------------------------------------------


constructor TTestManager.Create( aName : string; aWnd : HWND);
begin
  inherited Create;

  // Init everything
  FName   := aName;
  FHandle := aWnd;
end;


destructor TTestManager.Destroy;
begin
  try
    // add code here
  finally
    inherited Destroy;
  end;
end;


function TTestManager.Name : string;
begin
  result := FName;
end;


function TTestManager.Terminated : Boolean;
begin
  result := FTerminated;
end;


procedure TTestManager.Shutdown;
begin
  FTerminated := TRUE;
  while FRefCount > 1 do begin
    ReleaseSemaphore( g_hWakeUp, FRefCount, nil);
    g_evAkteur.WaitFor( 800);
  end;
end;


function TTestManager.CreateDbClient( aReset : Boolean = FALSE) : ITestDataDatabaseClient;
begin
  result := TTestDataDatabaseClient.Create( TESTDATA_DBSERVER_NAME, TESTDATA_DBSERVER_PORT);
  result.SetKeyspace( TESTDATA_KEYSPACE);
  if aReset then result.ReInitializeKeyspace;
end;


procedure TTestManager.NotifyErrors( iNr, iErrors : Integer);
begin
  PostMessage( FHandle, WMA_NOTIFYERRORS, WPARAM(iNr), LPARAM(iErrors));
end;


{
function TTestManager.NeueBestellung( const aClient : ITestDataDatabaseClient; aKundenNr : Integer) : IOrder;
begin
  if not Terminated
  then result := TOrder.Create( aClient, aKundenNr)
  else result := nil;
end;
}


//--- TPerson ------------------------------------------------------------------


constructor TPerson.Create( const aManager : ITestManager; aNr : Integer);
begin
  FManager    := aManager;
  FNr       := aNr;

  // negative und Nullnummern sind illegal
  ASSERT( aNr > 0);

  FreeOnTerminate := TRUE;
  inherited Create(FALSE);
end;


destructor TPerson.Destroy;
begin
  try
    FDbClient := nil;
    FManager    := nil;
    g_evAkteur.SetEvent;
  finally
    inherited Destroy;
  end;
end;


class function TPerson.CreateGUID : TGuid;
begin
  OleCheck( SysUtils.CreateGuid( result));
end;


procedure TPerson.Execute;
begin
  try
    FDbClient := FManager.CreateDbClient;
    while not (Terminated or FManager.Terminated) do begin
      try
        PerformTheTask;
        // TODO: consistency check after each task
      except
        on e:Exception do OutputDebugString( PChar( 'ERR: '+ClassName+'.PerformTask: '+e.Message));
      end;
      WaitForSingleObject( g_hWakeUp, FWaitFix + Random(Integer(FWaitRnd)));
    end;

  except
    on e:Exception do OutputDebugString( PChar( 'ERR: '+ClassName+': '+e.Message));
  end;
end;


//--- TReadWriteTester -------------------------------------------------------------------


constructor TReadWriteTester.Create( const aManager : ITestManager; aNr : Integer; aLevel : TConsistencyLevel);
begin
  FConsistency := aLevel;
  FErrors := 0;

  FWaitFix := 0;
  FWaitRnd := 0;
  inherited Create( aManager, aNr);
end;


procedure TReadWriteTester.PerformTheTask;
var pData  : ITestDataFamily;
    nValue : Integer;
    nTest  : Integer;
begin
  pData := FDBClient.TestData as ITestDataFamily;

  // read, increment, write
  nValue := ReadValue(pData);
  Inc(nValue);
  WriteValue( pData, nValue);

  // read and compare
  nTest := ReadValue(pData);
  if nValue <> nTest then begin
    Inc(FErrors);
    FManager.NotifyErrors( Nr, FErrors);
  end;
end;


function TReadWriteTester.ReadValue( const pData : ITestDataFamily) : Integer;
var nValue : Integer;
begin
  nValue := 0;  // Startwert
  pData.ReadData( PasCassa.MakeNamesList([ READWRITETEST_ROWKEY]),
                  PasCassa.MakeNamesList([ IntToStr(Nr)]),
                  function( const data : PasCassa.TSliceData) : Boolean
                  begin
                    data.GetColumnData( '', IntToStr(Nr), nValue, nValue);
                    result := FALSE;  // only one row with this rowkey anyway
                  end,
                  FConsistency);
  result := nValue;
end;


procedure TReadWriteTester.WriteValue( const pData : ITestDataFamily; aValue : Integer);
var pBag : PasCassa.IMutationBag;
    pRow : PasCassa.IRowMutations;
    pFam : PasCassa.IColumnFamilyMutations;
begin
  pBag := PasCassa.TMutationBag.Create( PasCassa.CurrentTimeStamp);
  pRow := pBag.Add( READWRITETEST_ROWKEY);
  pFam := pRow.Add( pData.Name);
  pFam.SetData( '', IntToStr(Nr), aValue);
  pData.BatchMutate( pBag, FConsistency);
end;





initialization
  g_hWakeUp  := CreateSemaphore( nil, 0, MAXINT, '');
  g_evAkteur := TEvent.Create( nil, FALSE, FALSE, '');
  Randomize;

finalization
  CloseHandle( g_hWakeUp);
  FreeAndNil( g_evAkteur);

end.

